package org.example;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class HotProductPvTopN {
    public static void main(String[] args) throws Exception {

        String sink_table_path = "sink_table";

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 数据源
        String source = "    " +
                "   CREATE TABLE source_table(\n" +
                "        `userid` STRING," +
                "        `itemid` STRING," +
                "        `categoryId` STRING,\n" +
                "        `timestamp` STRING,\n" +
                "        `behavior` STRING,\n" +
                "        windowEventTime AS TO_TIMESTAMP(CONVERT_TZ(`timestamp`, 'GMT+08:00', 'Asia/Shanghai')),\n" +
                "        WATERMARK FOR windowEventTime AS windowEventTime - INTERVAL '2' SECOND\n" +
                "    ) WITH (\n" +
                "        'connector' = 'filesystem',\n" +
                "        'path' = 'src/main/resources/userBehavior.csv',\n" +
                "        'format' = 'csv'\n" +
                "        )";

        // 输出
        String sink =
                "CREATE TABLE " + sink_table_path + "(" +
                        "   `itemid` STRING," +
                        "   `clickTimes` BIGINT" +
                        ") WITH ('connector' = 'print')";


        // 转换算子
        String transfrom = "" +
                "select * " +
//                "select url, clickTimes, row_num " +
                "from (" +
                "   SELECT itemid, window_start, window_end, COUNT(itemid) AS clickTimes,\n" +
                "       ROW_NUMBER() OVER(PARTITION BY window_start, window_end ORDER BY COUNT(itemid) DESC) AS row_num" +
                "   FROM TABLE (HOP( TABLE source_table, DESCRIPTOR(windowEventTime),INTERVAL '5' MINUTES,INTERVAL '1' HOUR))" +
                "   WHERE behavior = 'pv' " +
                "   GROUP BY itemid, window_start, window_end" +
                ")" +
                "where row_num<=5";


        // source
        tableEnv.executeSql(source);
        // sink
        tableEnv.executeSql(sink);
        // transfrom
        Table transTable = tableEnv.sqlQuery(transfrom);


//        transTable.executeInsert("sink_table");

        tableEnv.toChangelogStream(transTable).print();
        env.execute();

    }
}